package study.cdc.debezium.support.embedded;

import io.debezium.embedded.EmbeddedEngine;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;

//https://blog.51cto.com/u_6271727/2413021
@Slf4j
//@Configuration
public class EmbeddedRunner implements CommandLineRunner {

    @Autowired
    private io.debezium.config.Configuration config;

    @Override
    public void run(String... args) throws Exception {
        EmbeddedEngine engine = EmbeddedEngine.create()
                .using(config)
                .using((success, message, error) -> {
                    log.info(">>>>>>>>>>>>> cal");
                    // 强烈建议加上此部分的回调代码，方便查看错误信息
                    if (!success && error != null) {
                        // 报错回调
                        System.out.println("----------error------");
                        System.out.println(message);
                    }
                })
               // .notifying(new SourceRecordConsumer()::handle)
                //.notifying(new SourceRecordHandler()::handleBatch)
                .build();
        Thread t = new Thread(engine, "Embedded-Thread");
        t.start();

        addShutdownHook(engine);
    }

    private void addShutdownHook(EmbeddedEngine engine) {
        Runtime.getRuntime()
                .addShutdownHook(new Thread(() -> {
                    try {
                        log.info(">>>>>> close engine");
                        engine.close();
                    } catch (Exception ex) {
                    }
                }));
    }
}
